package com.gyj.server.cluster;

import com.gyj.server.entity.AbstractRequest;
import com.gyj.server.entity.CancelRequest;
import com.gyj.server.entity.HeartbeatRequest;
import com.gyj.server.entity.RegisterRequest;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.springframework.util.CollectionUtils;

import java.util.ArrayList;
import java.util.List;
import java.util.Objects;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * 集群相关
 * 2021/7/7
 * @author Gao
 */
public class PeersReplicator {

    private static final Logger logger = LogManager.getLogger(PeersReplicator.class);

    /**
     * 集群同步生成batch的间隔时间：500ms
     */
    private static final long PEERS_REPLICATE_BATCH_INTERVAL = 500;

    private static PeersReplicator instance = new PeersReplicator();

    private PeersReplicator() {
        // 启动接收请求和打包batch的线程
        AcceptorBatchQueueThread acceptorBatchQueueThread = new AcceptorBatchQueueThread();
        acceptorBatchQueueThread.setDaemon(true);
        acceptorBatchQueueThread.start();

        //启动同步发送batch的线程
        PeersReplicateThread peersReplicateThread = new PeersReplicateThread();
        peersReplicateThread.setDaemon(true);
        peersReplicateThread.start();
    }

    public static PeersReplicator getInstance() {
        return instance;
    }

    /**
     * 第一层队列：接收高并发的写入，无界队列，线程安全采用CAS机制
     * 不用LinkedBlocking有界队列原因：（eureka用的这个）
     * 1.LinkedBlockingQueue是用独占锁的方式的
     * 2.第一层队列是有界还是无界的，如果做成有界的，限制大小，万一有点什么意外，
     * 可能会导致在有界队列这里直接打满队列，会导致整个服务注册中心实例所有线程全部阻塞在这里，假死的状态。
     * 就经验而言，有界队列被打满的风险要高于无界队列内存被撑爆的风险，所以用LinkedBlockingQueue未必合适。
     */
    private ConcurrentLinkedQueue<AbstractRequest> acceptorQueue = new ConcurrentLinkedQueue<>();

    /**
     * 第二层队列负责打包队列成一个batch
     * LinkedBlockingQueue，一方面是独占锁，有界的，对第二个队列的入队和出队的操作都是少数的后台线程在执行，
     * 并不是大量线程高并发在写入的，有界的特性，避免说在异常情况下的过度内存消耗和膨胀，是合理的，是可行的。
     */
    private LinkedBlockingQueue<AbstractRequest> acceptorBatchQueue = new LinkedBlockingQueue<>(1000000);

    /**
     * 第三层队列：有界队列，用于batch的同步发送
     * LinkedBlockingQueue有界队列，添加元素三个方法，分别是add、put和offer
     * add 在添加元素的时候，若超出了度列的长度会直接抛出异常：
     * put 若向队尾添加元素的时候发现队列已经满了会发生阻塞一直等待空间，以加入元素。
     * offer 在添加元素时，如果发现队列已满无法添加的话，会直接返回false。
     * LinkedBlockingQueue的poll、remove跟take的区别：
     * poll: 若队列为空，返回null
     * remove:若队列为空，抛出NoSuchElementException异常。
     * take:若队列为空，发生阻塞，等待有元素。
     */
    private LinkedBlockingQueue<PeersReplicateBatch> peersReplicateQueue = new LinkedBlockingQueue<>(10000);

    /**
     * 同步服务注册请求
     */
    public void replicateRegister(RegisterRequest request) {
        acceptorQueue.offer(request);
    }

    /**
     * 同步服务下线请求
     */
    public void replicateCancel(CancelRequest request) {
        acceptorQueue.offer(request);
    }

    /**
     * 同步发送心跳请求
     */
    public void replicateHeartbeat(HeartbeatRequest request) {
        acceptorQueue.offer(request);
    }

    /**
     * 负责接收数据以及打包为batch的后台线程
     */
    private class AcceptorBatchQueueThread extends Thread {
        /**
         * 打包队列的时间
         */
        private long lastBatchTime = System.currentTimeMillis();

        @Override
        public void run() {
            while (true) {
                try {
                    // 从第一层队列获取请求，并且放入第二层队列中
                    AbstractRequest request = acceptorQueue.poll();
                    if (Objects.nonNull(request)) {
                        acceptorBatchQueue.put(request);
                    }
                    // 采用一定的策略来进行打包，每隔500ms生成一个batch
                    if (System.currentTimeMillis() - lastBatchTime > PEERS_REPLICATE_BATCH_INTERVAL) {
                        // 如果第二层队列里面有数据，全部打包生成一个batch放入第三个队列
                        if (acceptorBatchQueue.size() > 0) {
                            packBatchQueue();
                        }
                        lastBatchTime = System.currentTimeMillis();
                    }
                } catch (Exception e) {
                    logger.error("负责接收数据以及打包为batch的后台线程, 异常e:{}", e);
                }
            }
        }

        private PeersReplicateBatch packBatchQueue() {
            PeersReplicateBatch batch = new PeersReplicateBatch();
            acceptorQueue.forEach(batch::add);
            peersReplicateQueue.offer(batch);
            acceptorBatchQueue.clear();
            return batch;
        }
    }

    /**
     * 启动同步发送batch的线程
     */
    private class PeersReplicateThread extends Thread {
        /**
         * 自定义线程池
         */
        ExecutorService threadPool = new ThreadPoolExecutor(RegisterServerCluster.getPeers().size(),
                Integer.MAX_VALUE, 60L, TimeUnit.SECONDS,
                new LinkedBlockingQueue<Runnable>(100),
                // 自定义线程池的名称
                new ThreadFactory() {
            private AtomicInteger threadIndex = new AtomicInteger(0);
            @Override
            public Thread newThread(Runnable r) {
                return new Thread(r, "thread-" + this.threadIndex.incrementAndGet());
            }
        });

        @Override
        public void run() {
            while (true) {
                try {
                    PeersReplicateBatch batch = peersReplicateQueue.take();
                    if (batch != null) {
                        List<String> peers = RegisterServerCluster.getPeers();
                        if (!CollectionUtils.isEmpty(peers)) {
                            for(String peer : peers) {
                                threadPool.execute(new Runnable() {
                                    @Override
                                    public void run() {
                                        // 遍历所有的其他的register-server地址
                                        // 给每个地址的register-server都发送一个http请求同步batch
                                        System.out.println("给" + peer + "发送请求，同步batch过去......");
                                    }
                                });
                            }
                        }
                    }
                } catch (Exception e) {
                    logger.error("启动同步发送batch的线程, 异常e:{}", e);
                }
            }
        }
    }
}
